package com.yc.rpc.core.client;

import com.alibaba.fastjson.JSONObject;
import com.yc.rpc.core.Metadata;
import com.yc.rpc.core.RpcClientConfig;
import com.yc.rpc.core.server.ServerCommand;
import com.yc.rpc.core.server.ServerDomain;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.lang.reflect.Array;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.SynchronousQueue;

@Component
public class NettyClient implements InitializingBean {

    private volatile Channel channel;

    public static Map<String, HashMap<String, ServerDomain>> servers = new HashMap<>();
    public Map<String, Channel> channelMap = new HashMap<>();

    @Autowired
     private RpcClientConfig rpcClientConfig;
    @Autowired
    NettyClientHandler nettyClientHandler;

    @Value("${rpc.hundun.server}")
    private String rpcHunServer;

    Bootstrap b = new Bootstrap();

    public void connect(String host, int port) {
        EventLoopGroup worker = new NioEventLoopGroup();
        try {

            /**
             *EventLoop的组
             */
            b.group(worker);
            /**
             * 用于构造socketchannel工厂
             */
            b.channel(NioSocketChannel.class);
            /**设置选项
             * 参数：Socket的标准参数（key，value），可自行百度
             保持呼吸，不要断气！
             * */
            b.option(ChannelOption.SO_KEEPALIVE, true);
            /**
             * 自定义客户端Handle（客户端在这里搞事情）
             */
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch){
                    ch.pipeline().addLast(new LineBasedFrameDecoder(Integer.MAX_VALUE));
                    ch.pipeline().addLast(nettyClientHandler);
                }
            });
        } finally {

//            worker.shutdownGracefully();
        }
    }

    public Channel reconnect(String host, int port) {
        Channel channel = b.connect(host, port).channel();
        if (!channel.isActive()) {
            channel = reconnect(host, port);
        }
        return channel;
    }


    public SynchronousQueue<String> send(String serverId,Metadata msg) {

        HashMap<String, ServerDomain> stringServerDomainHashMap = servers.get(serverId);
        List<String> strings = new ArrayList<>(stringServerDomainHashMap.keySet());
        int num = (int) (Math.random() * strings.size());
        String address = strings.get(num);
        String key = serverId + address;
        Channel channel = channelMap.get(key);
        if (Objects.nonNull(channel)) {
            if (!channel.isActive()) {
                String[] split = address.split(":");
                channel = reconnect(split[0], Integer.parseInt(split[1]));
            }
            return nettyClientHandler.send(channel, msg);
        } else {
            String[] split = address.split(":");
            channel = reconnect(split[0], Integer.parseInt(split[1]));
            channelMap.put(key, channel);
            return nettyClientHandler.send(channel, msg);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        try {
            connect(rpcClientConfig.serverHost, rpcClientConfig.serverPort);
        }catch (Exception ignored){

        }

    }
}